package com.shujia.core

import org.apache.flink.streaming.api.scala._

object Demo04Parallelism {
  def main(args: Array[String]): Unit = {
    /**
     * Flink任务的并行度
     * 1、如果什么参数都不给 默认情况下任务只有一个并行度
     * 2、如果设置了全局并行度 除了一些特殊的任务只能有一个并行度外 大部分转换都会参考全局并行度
     * 3、DataStream可以单独设置并行度
     * 4、如果两个节点之间没有产生shuffle 在并行度一致的情况下可以合并
     * 5、Flink任务运行时slot的数量是由最大的并行度决定
     *
     *
     * 并行度的设置优先级：
     * 1、单独对每一个节点设置 wordsDS.setParallelism(3)
     * 2、通过Flink入口设置全局并行度 env.setParallelism(2)
     * 3、提交任务的时候可以在页面设置 或者 通过 -p 参数设置
     */

    /**
     * 创建Flink入口
     */

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    // 默认并行度等于CPU的逻辑核数 相当于是任务的一个并行度
    // 对任务设置全局的并行度
    //    env.setParallelism(2)

    /**
     * 通过Socket模拟实时数据
     * nc -lk 8888
     *
     * DataStream: Flink中的编程模型 类似Spark中的DStream
     * 通过Socket创建一个无界流
     *
     * Socket接收数据只能有一个并行度
     */
    val linesDS: DataStream[String] = env.socketTextStream("master", 8888)

    // 获取每个DataStream的并行度
    println(s"linesDS的并行度为：${linesDS.parallelism}")

    //    linesDS.print()

    // 对每一条数据进行切分
    val wordsDS: DataStream[String] = linesDS.flatMap(line => line.split(","))
    // 可以给每个DataStream单独设置并行度
    wordsDS.setParallelism(3)
    println(s"wordsDS的并行度为：${wordsDS.parallelism}")
    // 也可以个每个DataStream设置名称
    wordsDS.name("将每个单词切分")


    // 将每个单词变成 K V格式
    val wordsKVDS: DataStream[(String, Int)] = wordsDS.map(word => (word, 1))
    wordsKVDS.setParallelism(3)
    println(s"wordsKVDS的并行度为：${wordsKVDS.parallelism}")
    wordsKVDS.name("将每个单词变成kv格式")

    // 按照每个单词进行分组
    /**
     * KeyedStream不可以设置并行度 默认等于全局的并行度
     * 也不能设置名称 因为KeyedStream是隐藏节点
     */
    val keyByDS: KeyedStream[(String, Int), String] = wordsKVDS.keyBy(kv => kv._1)
    println(s"keyByDS的并行度为：${keyByDS.parallelism}")

    // 统计每个单词的数量 指定位置进行累加
    val wordCntDS: DataStream[(String, Int)] = keyByDS.sum(1)
    println(s"wordCntDS的并行度为：${wordCntDS.parallelism}")
    wordCntDS.name("统计单词数量")

    /**
     * Flink中的算子自带状态
     */
    // 将结果打印
    wordCntDS.print().setParallelism(1)


    // 启动任务
    env.execute("Demo04Parallelism")


  }

}
